package marsdonne.tasks;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.java.io.jdbc.JDBCOutputFormat;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

import java.io.Serializable;
import java.sql.Timestamp;
import java.text.SimpleDateFormat;

public class ForwardingTask implements Serializable {

    public enum DbType {
        PgSql("org.postgresql.Driver|jdbc:postgresql://192.168.0.198:5432/stat-examples?user=postgres&password=postgres"),
        H2("org.h2.Driver|jdbc:h2:./prodDb;MVCC=TRUE;LOCK_TIMEOUT=10000;DB_CLOSE_ON_EXIT=FALSE");

        private final String driver;

        DbType(String driver) {
            this.driver = driver;
        }
    }


    private final JDBCOutputFormat jdbcOutput;
    private final int pCounter;

    public ForwardingTask(DbType dbType, String insertSQL, int[] typesArray) {
        final String[] dbConfig = dbType.driver.split("\\|");

        this.jdbcOutput = JDBCOutputFormat.buildJDBCOutputFormat()
                .setDrivername(dbConfig[0])
                .setDBUrl(dbConfig[1])
                .setQuery(insertSQL)
                .setSqlTypes(typesArray)
                .setBatchInterval(1)//默认批量太大，不方便测试。
                .finish();

        this.pCounter = typesArray.length;
    }

    public void submitTo(StreamExecutionEnvironment env, SourceFunction<String> source) throws Exception {
        final SimpleDateFormat simpleDateFormattor = new SimpleDateFormat("yyyy-MM-dd");

        env.addSource(source)
                .returns(String.class)//兼容测试环境 “集合源”
                .setParallelism(1)                 // non-parallel source is only required for exactly-once
                .flatMap((String dataJSON, Collector<Row> collector) -> {
                    JSONObject jsonData = JSON.parseObject(dataJSON);//decoding json

                    Row row = new Row(this.pCounter);

                    row.setField(0, jsonData.getString("type"));//set type
                    row.setField(1, new Timestamp(simpleDateFormattor.parse(jsonData.getString("timed")).getTime()));//set timed
                    row.setField(2, jsonData.getJSONObject("metadata").toJSONString());//set metadata
                    row.setField(3, jsonData.getJSONObject("contents").toJSONString());//set contents

                    collector.collect(row);
                })
                .returns(Row.class)
                .writeUsingOutputFormat(jdbcOutput);

        // execute program
        env.execute();
    }
}
